/**
 * @file SRTest.cpp
 *
 * This file implements the test for the StreamReader.
 *
 */

#include "SRTest.h"

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <string.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>

#include "cute.h"
#include "StreamReader.h"
#include "StreamAnalyser.h"
#include "TS.h"

#define MAX_BUFF_SIZE   1500
#define MAX_CB_ELT 1000
#define PACKET_PER_BUFFER 7 //  7 max


// Test functions

/**
 * Test StreamReader Creation/Deletion for UDP input
 *
 */
void SRTest_CreateDeleteUdp()
{
#define WAIT_TIME 100000
#define PORT	9400
#define ADDR	"127.0.0.1"
#define BIND_ADDR	"0.0.0.0"

    SR_Option_t SROptions;
    SR_StreamReader_t * TestSR_p = NULL;
	int Error = OK;

	bzero((void*)&SROptions, sizeof(SR_Option_t));
	Error = CircB_CreateCircBuff(&SROptions.CB, MAX_BUFF_SIZE+sizeof(Tag_t), MAX_CB_ELT);
	ASSERT_EQUALM("Error", 0, Error);
 	gettimeofday(&SROptions.ReferenceTime, NULL);
 	SROptions.InputType = Udp;
 	strcpy(SROptions.BindAddress, BIND_ADDR);
 	strcpy(SROptions.RcvAddress, ADDR);
 	SROptions.RcvPort = PORT;

 	// Test creation
 	Error = SR_Create(&TestSR_p, SROptions);

	ASSERT_EQUALM("Error", 0, Error);
	ASSERTM("TestSR_p is NULL", TestSR_p != NULL);
	ASSERT_EQUALM("TestSR_p->Options.InputType", (int)Udp, (int)TestSR_p->Options.InputType);
	ASSERT_EQUALM("TestSR_p->Options.RcvPort", PORT, TestSR_p->Options.RcvPort);
	Error = strcmp(BIND_ADDR, TestSR_p->Options.BindAddress);
	ASSERT_EQUALM("TestSR_p->Options.BindAddress", 0, Error);
	Error = strcmp(ADDR, TestSR_p->Options.RcvAddress);
	ASSERT_EQUALM("TestSR_p->Options.RcvAddress", 0, Error);
	ASSERT_EQUALM("TestSR_p->Options.FileName", 0, TestSR_p->Options.FileName[0]);
	ASSERTM("TestSR_p->Options.CB is NULL", TestSR_p->Options.CB != NULL);
	ASSERT_EQUALM("TestSR_p->ThreadLoop", 1, TestSR_p->ThreadLoop);
	ASSERTM("TestSR_p->InputFile is not NULL", TestSR_p->InputFile == NULL);

	printf("Waiting %i µs...\n", WAIT_TIME);
	usleep(WAIT_TIME);

	ASSERTM("TestSR_p->Socket is 0", TestSR_p->Socket != 0);

    // Delete
	Error = SR_Delete(TestSR_p);
	ASSERT_EQUALM("Error", 0, Error);

	Error = CircB_DeleteCircBuff(SROptions.CB);
	ASSERT_EQUALM("Error", 0, Error);

#undef WAIT_TIME
#undef PORT
#undef ADDR
#undef BIND_ADDR
}

/**
 * Test StreamReader Creation/Deletion for TCP input
 *
 */
void SRTest_CreateDeleteTcp()
{
#define WAIT_TIME 100000
#define PORT	9400
#define ADDR	"127.0.0.1"
#define BIND_ADDR	"0.0.0.0"

    SR_Option_t SROptions;
    SR_StreamReader_t * TestSR_p = NULL;
	int Error = OK;

	bzero((void*)&SROptions, sizeof(SR_Option_t));
	Error = CircB_CreateCircBuff(&SROptions.CB, MAX_BUFF_SIZE+sizeof(Tag_t), MAX_CB_ELT);
	ASSERT_EQUALM("Error", 0, Error);
 	gettimeofday(&SROptions.ReferenceTime, NULL);
 	SROptions.InputType = Tcp;
 	strcpy(SROptions.BindAddress, BIND_ADDR);
 	strcpy(SROptions.RcvAddress, ADDR);
 	SROptions.RcvPort = PORT;

 	// Test creation
 	Error = SR_Create(&TestSR_p, SROptions);

	ASSERT_EQUALM("Error", 0, Error);
	ASSERTM("TestSR_p is NULL", TestSR_p != NULL);
	ASSERT_EQUALM("TestSR_p->Options.InputType", (int)Tcp, (int)TestSR_p->Options.InputType);
	ASSERT_EQUALM("TestSR_p->Options.RcvPort", PORT, TestSR_p->Options.RcvPort);
	Error = strcmp(BIND_ADDR, TestSR_p->Options.BindAddress);
	ASSERT_EQUALM("TestSR_p->Options.BindAddress", 0, Error);
	Error = strcmp(ADDR, TestSR_p->Options.RcvAddress);
	ASSERT_EQUALM("TestSR_p->Options.RcvAddress", 0, Error);
	ASSERT_EQUALM("TestSR_p->Options.FileName", 0, TestSR_p->Options.FileName[0]);
	ASSERTM("TestSR_p->Options.CB is NULL", TestSR_p->Options.CB != NULL);
	ASSERT_EQUALM("TestSR_p->ThreadLoop", 1, TestSR_p->ThreadLoop);
	ASSERTM("TestSR_p->InputFile is not NULL", TestSR_p->InputFile == NULL);

	printf("Waiting %i µs...\n", WAIT_TIME);
	usleep(WAIT_TIME);

	ASSERTM("TestSR_p->ListenSocket is 0", TestSR_p->ListenSocket != 0);

	// Delete
	Error = SR_Delete(TestSR_p);
	ASSERT_EQUALM("Error", 0, Error);

	Error = CircB_DeleteCircBuff(SROptions.CB);
	ASSERT_EQUALM("Error", 0, Error);

#undef WAIT_TIME
#undef PORT
#undef ADDR
#undef BIND_ADDR
}


/**
 * Test StreamReader Creation/Deletion for File input
 *
 */
void SRTest_CreateDeleteFile()
{
#define WAIT_TIME 100000
#define FILENAME "test34.ts"

    SR_Option_t Options;
    SR_StreamReader_t * TestSR_p = NULL;
	int Error = OK;

	bzero((void*)&Options, sizeof(SR_Option_t));
	Error = CircB_CreateCircBuff(&Options.CB, MAX_BUFF_SIZE+sizeof(Tag_t), MAX_CB_ELT);
	ASSERT_EQUALM("Error", 0, Error);
 	gettimeofday(&Options.ReferenceTime, NULL);
 	Options.InputType = File;
 	strcpy(Options.FileName, FILENAME);

 	// Test creation
 	Error = SR_Create(&TestSR_p, Options);

	ASSERT_EQUALM("Error", 0, Error);
	ASSERTM("TestSR_p is NULL", TestSR_p != NULL);
	ASSERT_EQUALM("TestSR_p->Options.InputType", (int)File, (int)TestSR_p->Options.InputType);
	ASSERT_EQUALM("TestSR_p->Options.RcvPort", 0, TestSR_p->Options.RcvPort);
	Error = strcmp(FILENAME, TestSR_p->Options.FileName);
	ASSERTM("TestSR_p->Options.CB is NULL", TestSR_p->Options.CB != NULL);
	ASSERT_EQUALM("TestSR_p->ThreadLoop", 1, TestSR_p->ThreadLoop);

	printf("Waiting %i µs...\n", WAIT_TIME);
	usleep(WAIT_TIME);

	ASSERTM("TestSR_p->InputFile is NULL", TestSR_p->InputFile != NULL);

	// Delete
	Error = SR_Delete(TestSR_p);
	ASSERT_EQUALM("Error", 0, Error);

	Error = CircB_DeleteCircBuff(Options.CB);
	ASSERT_EQUALM("Error", 0, Error);

#undef WAIT_TIME
#undef FILENAME
}


/**
 * Test StreamReader for UDP input
 *
 */
void SRTest_ReceiveUdp()
{
#define WAIT_TIME 100000
#define INPUT_LENGTH 50
#define PORT	9400
#define ADDR	"127.0.0.1"
#define BIND_ADDR	"0.0.0.0"

    SR_Option_t SROptions;
    SR_StreamReader_t * TestSR_p = NULL;

	int ReadCount = 1;
    unsigned char Buffer[PACKET_PER_BUFFER*TS_SIZE];
	struct stat s;
	unsigned long int FilePacketNumber = 0;
	char InputName[INPUT_LENGTH];
	FILE * TestFile = NULL;
	void * Buff_p = NULL;
	unsigned int Size = 0;
	int Error = OK;

	int SendCount = 0;
	int SendSock = 0;
	struct sockaddr_in RemoteAddr;

	for (int i=34; i<=37; i++)
	{
		bzero((void*)&SROptions, sizeof(SR_Option_t));
		Error = CircB_CreateCircBuff(&SROptions.CB, MAX_BUFF_SIZE+sizeof(Tag_t), MAX_CB_ELT);
		ASSERT_EQUALM("Error", 0, Error);
	 	gettimeofday(&SROptions.ReferenceTime, NULL);
	 	SROptions.InputType = Udp;
	 	strcpy(SROptions.BindAddress, BIND_ADDR);
	 	strcpy(SROptions.RcvAddress, ADDR);
	 	SROptions.RcvPort = PORT;

	 	// Create SR
	 	Error = SR_Create(&TestSR_p, SROptions);

		ASSERT_EQUALM("Error", 0, Error);
		ASSERTM("TestSR_p is NULL", TestSR_p != NULL);

		printf("Waiting %i µs...\n", WAIT_TIME);
		usleep(WAIT_TIME);

		// Open the file to send
		snprintf(InputName, INPUT_LENGTH, "test%i.ts", i);
		TestFile = fopen(InputName, "r");
		ASSERTM("TestFile is NULL", TestFile != NULL);

		printf("Testing UDP with file: %s\n", InputName);
		stat(InputName, &s);
		FilePacketNumber = s.st_size/TS_SIZE;
		printf ("File size:  %li B => %li TS packets\n", s.st_size, FilePacketNumber);

	    // Open the socket
		SendSock = socket(AF_INET, SOCK_DGRAM, 0);
	    ASSERTM("SendSock", SendSock > 0);

	    unsigned int opt = 1;
	    Error = setsockopt(SendSock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
		ASSERT_EQUALM("Error", 0, Error);

		bzero((char *) &RemoteAddr, sizeof(RemoteAddr));
		RemoteAddr.sin_family = AF_INET;
		RemoteAddr.sin_addr.s_addr = inet_addr(ADDR);
		RemoteAddr.sin_port = htons(PORT);

		// Send file through the socket
		do
		{
			bzero((void *)Buffer, PACKET_PER_BUFFER*TS_SIZE);
			ReadCount = fread((void*)Buffer, 1, PACKET_PER_BUFFER*TS_SIZE, TestFile);
			if (ReadCount > 0)
			{
			    SendCount = sendto(SendSock, (void *)Buffer, ReadCount, 0, (struct sockaddr*) &RemoteAddr, (socklen_t) sizeof(struct sockaddr_in));
			}

			if ((ReadCount <= 0) || (SendCount <= 0) || (feof(TestFile)))
			{
				// Error
				break;
			}
		}
		while ((ReadCount > 0) && (SendCount > 0) && (!feof(TestFile)));

		printf("Waiting 1 s...\n");
		sleep(1);

	    shutdown(SendSock, SHUT_RDWR);
	    close(SendSock);

		if (TestFile != NULL)
		{
			fclose(TestFile);
		}

		// Verify reception
		ASSERT_EQUALM("TestSR_p->TotalRxCount", s.st_size, TestSR_p->TotalByteRead);

		printf("CircBuff content: \n");
		unsigned int Index = 0;
		while (Index != TestSR_p->Options.CB->WriteIndex)
		{
			Error = CircB_Watch(TestSR_p->Options.CB, Index, &Buff_p, &Size);
			if ((Error == CIRCB_WARNING_END_OF_BUFF) || (Error == CIRCB_WARNING_BEG_OF_BUFF))
			{
				Error = OK;
			}
			ASSERT_EQUALM("Error", 0, Error);

			printf("%i\n", Size-sizeof(Tag_t));

			ASSERT_EQUALM("Tag->ReadyToPlay", 0, ((Tag_t*)Buff_p)->ReadyToPlay);
			ASSERTM("Tag->ReceptionTime.tv_usec", (((Tag_t*)Buff_p)->ReceptionTime.tv_usec) != 0);
			ASSERT_EQUALM("DVB Sync byte", 0x47, (int)*(((unsigned char*)Buff_p)+sizeof(Tag_t)));

			Index++;
			Index %= TestSR_p->Options.CB->MaxElementNumber;
		}

	    // Delete
		Error = SR_Delete(TestSR_p);
		ASSERT_EQUALM("Error", 0, Error);

		Error = CircB_DeleteCircBuff(SROptions.CB);
		ASSERT_EQUALM("Error", 0, Error);
	}

#undef WAIT_TIME
#undef INPUT_LENGTH
#undef PORT
#undef ADDR
#undef BIND_ADDR
}

/**
 * Test StreamReader for TCP input
 *
 */
void SRTest_ReceiveTcp()
{
#define WAIT_TIME 100000
#define INPUT_LENGTH 50
#define PORT	9400
#define ADDR	"127.0.0.1"
#define BIND_ADDR	"0.0.0.0"

    SR_Option_t SROptions;
    SR_StreamReader_t * TestSR_p = NULL;

	int ReadCount = 1;
    unsigned char Buffer[PACKET_PER_BUFFER*TS_SIZE];
	struct stat s;
	unsigned long int FilePacketNumber = 0;
	char InputName[INPUT_LENGTH];
	FILE * TestFile = NULL;
	void * Buff_p = NULL;
	unsigned int Size = 0;
	int Error = OK;

	int SendCount = 0;
	int SendSock = 0;
	struct sockaddr_in RemoteAddr;

	for (int i=34; i<=37; i++)
	{
		bzero((void*)&SROptions, sizeof(SR_Option_t));
		Error = CircB_CreateCircBuff(&SROptions.CB, MAX_BUFF_SIZE+sizeof(Tag_t), MAX_CB_ELT);
		ASSERT_EQUALM("Error", 0, Error);
	 	gettimeofday(&SROptions.ReferenceTime, NULL);
	 	SROptions.InputType = Tcp;
	 	strcpy(SROptions.BindAddress, BIND_ADDR);
	 	strcpy(SROptions.RcvAddress, ADDR);
	 	SROptions.RcvPort = PORT;

	 	Error = SR_Create(&TestSR_p, SROptions);

		ASSERT_EQUALM("Error", 0, Error);
		ASSERTM("TestSR_p is NULL", TestSR_p != NULL);

		printf("Waiting %i µs...\n", WAIT_TIME);
		usleep(WAIT_TIME);

		snprintf(InputName, INPUT_LENGTH, "test%i.ts", i);
		TestFile = fopen(InputName, "r");
		ASSERTM("TestFile is NULL", TestFile != NULL);

		printf("Testing TCP with file: %s\n", InputName);
		stat(InputName, &s);
		FilePacketNumber = s.st_size/TS_SIZE;
		printf ("File size:  %li B => %li TS packets\n", s.st_size, FilePacketNumber);

	    SendSock = socket(AF_INET, SOCK_STREAM, 0);
	    ASSERTM("SendSock", SendSock > 0);

	    unsigned int opt = 1;
	    Error = setsockopt(SendSock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
		ASSERT_EQUALM("Error", 0, Error);

		bzero((char *) &RemoteAddr, sizeof(RemoteAddr));
		RemoteAddr.sin_family = AF_INET;
		RemoteAddr.sin_addr.s_addr = inet_addr(ADDR);
		RemoteAddr.sin_port = htons(PORT);

		Error = connect(SendSock, (struct sockaddr *) &RemoteAddr, sizeof(RemoteAddr));
		ASSERT_EQUALM("Error", 0, Error);

		// Transmit
		do
		{
			bzero((void *)Buffer, PACKET_PER_BUFFER*TS_SIZE);
			ReadCount = fread((void*)Buffer, 1, PACKET_PER_BUFFER*TS_SIZE, TestFile);
			if (ReadCount > 0)
			{
			    SendCount = send(SendSock, (void *)Buffer, ReadCount, 0);
			}

			if ((ReadCount <= 0) || (SendCount <= 0) || (feof(TestFile)))
			{
				// Error
				break;
			}
		}
		while ((ReadCount > 0) && (SendCount > 0) && (!feof(TestFile)));

		printf("Waiting 1.5 s...\n");
		usleep(1500000);

	    shutdown(SendSock, SHUT_RDWR);
	    close(SendSock);

		if (TestFile != NULL)
		{
			fclose(TestFile);
		}

		// Check reception
		ASSERT_EQUALM("TestSR_p->TotalRxCount", s.st_size, TestSR_p->TotalByteRead);

		printf("CircBuff content: \n");
		unsigned int Index = 0;
		while (Index != TestSR_p->Options.CB->WriteIndex)
		{
			Error = CircB_Watch(TestSR_p->Options.CB, Index, &Buff_p, &Size);
			if ((Error == CIRCB_WARNING_END_OF_BUFF) || (Error == CIRCB_WARNING_BEG_OF_BUFF))
			{
				Error = OK;
			}
			ASSERT_EQUALM("Error", 0, Error);

			printf("%i\n", Size-sizeof(Tag_t));

			ASSERT_EQUALM("Tag->ReadyToPlay", 0, ((Tag_t*)Buff_p)->ReadyToPlay);
			ASSERTM("Tag->ReceptionTime.tv_usec", (((Tag_t*)Buff_p)->ReceptionTime.tv_usec) != 0);
			ASSERT_EQUALM("DVB Sync byte", 0x47, (int)*(((unsigned char*)Buff_p)+sizeof(Tag_t)));

			Index++;
			Index %= TestSR_p->Options.CB->MaxElementNumber;
		}

	    // Delete
		Error = SR_Delete(TestSR_p);
		ASSERT_EQUALM("Error", 0, Error);

		Error = CircB_DeleteCircBuff(SROptions.CB);
		ASSERT_EQUALM("Error", 0, Error);
	}

#undef WAIT_TIME
#undef INPUT_LENGTH
#undef PORT
#undef ADDR
#undef BIND_ADDR
}


/**
 * Test StreamReader for file input
 *
 */
void SRTest_ReadFile()
{
#define WAIT_TIME 1
#define INPUT_LENGTH 50

    SR_Option_t SROptions;
    SR_StreamReader_t * TestSR_p = NULL;
	struct stat s;
	unsigned long int FilePacketNumber = 0;
	char InputName[INPUT_LENGTH];
	int Error = OK;
	void * Buff_p = NULL;
	unsigned int Size = 0;

	for (int i=34; i<=37; i++)
	{
		bzero((void*)&SROptions, sizeof(SR_Option_t));
		Error = CircB_CreateCircBuff(&SROptions.CB, MAX_BUFF_SIZE+sizeof(Tag_t), MAX_CB_ELT);
		ASSERT_EQUALM("Error", 0, Error);

	 	SROptions.InputType = File;
		snprintf(InputName, INPUT_LENGTH, "test%i.ts", i);
	 	strcpy(SROptions.FileName, InputName);
	 	gettimeofday(&SROptions.ReferenceTime, NULL);

	 	// Test creation
		printf("Testing file: %s\n", InputName);
		stat(InputName, &s);
		FilePacketNumber = s.st_size/TS_SIZE;
		printf ("File size:  %li B => %li TS packets\n", s.st_size, FilePacketNumber);

	 	Error = SR_Create(&TestSR_p, SROptions);

		ASSERT_EQUALM("Error", 0, Error);
		ASSERTM("TestSR_p is NULL", TestSR_p != NULL);
		ASSERT_EQUALM("TestSR_p->Options.InputType", (int)File, (int)TestSR_p->Options.InputType);
		Error = strcmp(InputName, TestSR_p->Options.FileName);
		ASSERT_EQUALM("TestSR_p->Options.FileName", 0, Error);
		ASSERTM("TestSR_p->Options.CB is NULL", TestSR_p->Options.CB != NULL);
		ASSERT_EQUALM("TestSR_p->ThreadLoop", 1, TestSR_p->ThreadLoop);
		ASSERTM("TestSR_p->InputFile is NULL", TestSR_p->InputFile != NULL);

		printf("Waiting %i s...\n", WAIT_TIME);
		sleep(WAIT_TIME);

		int NumberOfChunk = FilePacketNumber/MAX_PACKET_PER_MTU + 1; // +1 because we add a null packet at the end of the file

		ASSERT_EQUALM("NumberOfChunk", NumberOfChunk, TestSR_p->Options.CB->ElementCount);

		unsigned int Index = 0;
		while (Index != TestSR_p->Options.CB->WriteIndex)
		{
			Error = CircB_Watch(TestSR_p->Options.CB, Index, &Buff_p, &Size);
			if ((Error == CIRCB_WARNING_END_OF_BUFF) || (Error == CIRCB_WARNING_BEG_OF_BUFF))
			{
				Error = OK;
			}
			ASSERT_EQUALM("Error", 0, Error);
			ASSERT_EQUALM("Tag->ReadyToPlay", 0, ((Tag_t*)Buff_p)->ReadyToPlay);
			ASSERT_EQUALM("DVB Sync byte", 0x47, (int)*(((unsigned char*)Buff_p)+sizeof(Tag_t)));

			// Verify that last chunk is marked properly
			if (Index == (TestSR_p->Options.CB->WriteIndex-1))
			{
				ASSERT_EQUALM("Tag->ChunkIsLast", 1, ((Tag_t*)Buff_p)->ChunkIsLast);
			}

			Index++;
			Index %= TestSR_p->Options.CB->MaxElementNumber;
		}

		// Delete
		Error = SR_Delete(TestSR_p);
		ASSERT_EQUALM("Error", 0, Error);

		Error = CircB_DeleteCircBuff(SROptions.CB);
		ASSERT_EQUALM("Error", 0, Error);
	}


#undef WAIT_TIME
#undef INPUT_LENGTH
}


